package com.by.rpc.transaction;

import com.alibaba.fastjson2.JSON;
import com.by.log.Logger;
import com.by.log.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 事务消息监听器
 * 处理异步事务消息
 */
@Component
public class TransactionMessageListener {
    // 添加自定义日志实例
    private static final Logger log = LoggerFactory.getLogger(TransactionMessageListener.class);
    
    private final RedisTemplate<String, Object> redisTemplate;
    private final RedisAtomicTransactionManager transactionManager;
    
    // 事务准备队列
    private static final String TRANSACTION_PREPARE_QUEUE = "atomic_transaction:prepare";
    
    // 事务提交队列
    private static final String TRANSACTION_COMMIT_QUEUE = "atomic_transaction:commit";
    
    // 事务回滚队列
    private static final String TRANSACTION_ROLLBACK_QUEUE = "atomic_transaction:rollback";
    
    // 消息处理线程池
    private ExecutorService executorService;
    
    // 监听器运行状态
    private volatile boolean running = true;
    
    public TransactionMessageListener(RedisTemplate<String, Object> redisTemplate, 
                                  RedisAtomicTransactionManager transactionManager) {
        this.redisTemplate = redisTemplate;
        this.transactionManager = transactionManager;
    }
    
    @PostConstruct
    public void init() {
        // 初始化线程池
        executorService = Executors.newFixedThreadPool(10);
        
        // 启动消息监听线程
        startListening();
        
        log.info("Transaction message listener initialized");
    }
    
    @PreDestroy
    public void destroy() {
        running = false;
        
        if (executorService != null) {
            executorService.shutdown();
            try {
                if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                    executorService.shutdownNow();
                }
            } catch (InterruptedException e) {
                executorService.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        
        log.info("Transaction message listener destroyed");
    }
    
    /**
     * 启动消息监听
     */
    private void startListening() {
        // 启动准备消息监听线程
        executorService.submit(this::listenPrepareMessages);
        
        // 启动提交消息监听线程
        executorService.submit(this::listenCommitMessages);
        
        // 启动回滚消息监听线程
        executorService.submit(this::listenRollbackMessages);
    }
    
    /**
     * 监听准备消息
     */
    private void listenPrepareMessages() {
        while (running) {
            try {
                // 从准备队列中获取消息
                Object messageObj = redisTemplate.opsForList().rightPop(TRANSACTION_PREPARE_QUEUE, 1, TimeUnit.SECONDS);
                if (messageObj != null) {
                    String messageJson = (String) messageObj;
                    RedisAtomicTransactionManager.TransactionMessage message = 
                            JSON.parseObject(messageJson, RedisAtomicTransactionManager.TransactionMessage.class);
                    
                    // 异步处理准备消息
                    executorService.submit(() -> processPrepareMessage(message));
                }
            } catch (Exception e) {
                log.error("Error while listening prepare messages", e);
                
                // 防止异常导致线程终止
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    
    /**
     * 监听提交消息
     */
    private void listenCommitMessages() {
        while (running) {
            try {
                // 从提交队列中获取消息
                Object messageObj = redisTemplate.opsForList().rightPop(TRANSACTION_COMMIT_QUEUE, 1, TimeUnit.SECONDS);
                if (messageObj != null) {
                    String messageJson = (String) messageObj;
                    RedisAtomicTransactionManager.TransactionMessage message = 
                            JSON.parseObject(messageJson, RedisAtomicTransactionManager.TransactionMessage.class);
                    
                    // 异步处理提交消息
                    executorService.submit(() -> processCommitMessage(message));
                }
            } catch (Exception e) {
                log.error("Error while listening commit messages", e);
                
                // 防止异常导致线程终止
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    
    /**
     * 监听回滚消息
     */
    private void listenRollbackMessages() {
        while (running) {
            try {
                // 从回滚队列中获取消息
                Object messageObj = redisTemplate.opsForList().rightPop(TRANSACTION_ROLLBACK_QUEUE, 1, TimeUnit.SECONDS);
                if (messageObj != null) {
                    String messageJson = (String) messageObj;
                    RedisAtomicTransactionManager.TransactionMessage message = 
                            JSON.parseObject(messageJson, RedisAtomicTransactionManager.TransactionMessage.class);
                    
                    // 异步处理回滚消息
                    executorService.submit(() -> processRollbackMessage(message));
                }
            } catch (Exception e) {
                log.error("Error while listening rollback messages", e);
                
                // 防止异常导致线程终止
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    
    /**
     * 处理准备消息
     * @param message 事务消息
     */
    private void processPrepareMessage(RedisAtomicTransactionManager.TransactionMessage message) {
        try {
            log.info("Processing prepare message for transaction: {}", message.getTransactionId());
            
            // 这里可以执行一些预处理操作
            RedisAtomicTransactionManager.TransactionContent content = message.getContent();
            if (content != null) {
                log.info("Transaction content - BusinessId: {}, OperationType: {}", 
                        content.getBusinessId(), content.getOperationType());
                
                // 根据业务类型执行不同的预处理逻辑
                // 例如：预扣款、预占资源等
            }
            
            log.info("Prepare message processed for transaction: {}", message.getTransactionId());
        } catch (Exception e) {
            log.error("Failed to process prepare message for transaction: {}", message.getTransactionId(), e);
        }
    }
    
    /**
     * 处理提交消息
     * @param message 事务消息
     */
    private void processCommitMessage(RedisAtomicTransactionManager.TransactionMessage message) {
        try {
            log.info("Processing commit message for transaction: {}", message.getTransactionId());
            
            // 执行真正的业务提交操作
            RedisAtomicTransactionManager.TransactionContent content = message.getContent();
            if (content != null) {
                log.info("Committing transaction content - BusinessId: {}, OperationType: {}", 
                        content.getBusinessId(), content.getOperationType());
                
                // 根据业务类型执行不同的提交逻辑
                // 例如：创建订单、更新库存、发送通知等
                handleBusinessCommit(content);
            }
            
            log.info("Commit message processed for transaction: {}", message.getTransactionId());
        } catch (Exception e) {
            log.error("Failed to process commit message for transaction: {}", message.getTransactionId(), e);
            
            // 提交失败时可以记录日志或发送告警
            handleCommitFailure(message);
        }
    }
    
    /**
     * 处理回滚消息
     * @param message 事务消息
     */
    private void processRollbackMessage(RedisAtomicTransactionManager.TransactionMessage message) {
        try {
            log.info("Processing rollback message for transaction: {}", message.getTransactionId());
            
            // 执行补偿操作
            RedisAtomicTransactionManager.TransactionContent content = message.getContent();
            if (content != null) {
                log.info("Rolling back transaction content - BusinessId: {}, OperationType: {}", 
                        content.getBusinessId(), content.getOperationType());
                
                // 根据业务类型执行不同的补偿逻辑
                // 例如：取消订单、恢复库存、退款等
                handleBusinessRollback(content);
            }
            
            log.info("Rollback message processed for transaction: {}", message.getTransactionId());
        } catch (Exception e) {
            log.error("Failed to process rollback message for transaction: {}", message.getTransactionId(), e);
            
            // 回滚失败时可以记录日志或发送告警
            handleRollbackFailure(message);
        }
    }
    
    /**
     * 处理业务提交
     * @param content 事务内容
     */
    private void handleBusinessCommit(RedisAtomicTransactionManager.TransactionContent content) {
        // 根据业务类型执行不同的提交逻辑
        String operationType = content.getOperationType();
        if ("SECKILL".equals(operationType)) {
            // 处理秒杀业务提交
            log.info("Handling seckill commit for business: {}", content.getBusinessId());
            
            // 这里可以执行数据库操作，如创建订单等
            // 由于在秒杀场景中，库存已经在Redis中扣减了，这里主要是持久化到数据库
        } else if ("ORDER".equals(operationType)) {
            // 处理订单业务提交
            log.info("Handling order commit for business: {}", content.getBusinessId());
        }
        
        // 其他业务类型的处理...
    }
    
    /**
     * 处理业务回滚
     * @param content 事务内容
     */
    private void handleBusinessRollback(RedisAtomicTransactionManager.TransactionContent content) {
        // 根据业务类型执行不同的补偿逻辑
        String operationType = content.getOperationType();
        if ("SECKILL".equals(operationType)) {
            // 处理秒杀业务回滚
            log.info("Handling seckill rollback for business: {}", content.getBusinessId());
            
            // 补偿操作，如恢复库存等
        } else if ("ORDER".equals(operationType)) {
            // 处理订单业务回滚
            log.info("Handling order rollback for business: {}", content.getBusinessId());
        }
        
        // 其他业务类型的处理...
    }
    
    /**
     * 处理提交失败
     * @param message 事务消息
     */
    private void handleCommitFailure(RedisAtomicTransactionManager.TransactionMessage message) {
        // 记录提交失败日志
        log.error("Transaction commit failed: {}", message.getTransactionId());
        
        // 可以发送告警通知
        // 可以将失败的事务信息存储到专门的失败队列中，供人工处理
    }
    
    /**
     * 处理回滚失败
     * @param message 事务消息
     */
    private void handleRollbackFailure(RedisAtomicTransactionManager.TransactionMessage message) {
        // 记录回滚失败日志
        log.error("Transaction rollback failed: {}", message.getTransactionId());
        
        // 可以发送告警通知
        // 可以将失败的事务信息存储到专门的失败队列中，供人工处理
    }
    
    /**
     * 定时检查未完成的事务
     */
    @Scheduled(fixedDelay = 30000) // 每30秒检查一次
    public void checkUnfinishedTransactions() {
        try {
            log.debug("Checking unfinished transactions");
            
            // 这里可以检查Redis中是否存在超时的事务
            // 并进行相应的处理
            
        } catch (Exception e) {
            log.error("Error while checking unfinished transactions", e);
        }
    }
}